package com.example.demo;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;

public class HbaseApiOperation {

    // 统一表名字
    private  final static String tableNameStr="hujinhui:student" ;
    public static final String ZK_CONNECT_VALUE = "47.101.204.23:2181,47.101.216.12:2181,47.101.206.249:2181";

    // 建立连接
    public static Connection getConnection() {
        try {
            //获取配置
            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", ZK_CONNECT_VALUE);
            Connection connection = ConnectionFactory.createConnection(config);
            return connection;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 创建表 ， 指定多个列族
     * @throws IOException
     * @param connection
     * @param tableName
     * @param columnFamilys
     * @param versions
     */
    public void createTable(Connection connection, TableName tableName, String[] columnFamilys, int versions) throws IOException {
        // 指定表名   表名：$your_name:student
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if (admin.tableExists(tableName)) {
                System.out.printf("table %s exists!", tableNameStr);
            } else {
                TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
                for (String columnFamily : columnFamilys) {
                    builder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(columnFamily));
                }
                admin.createTable(builder.build());
                System.out.printf("create table %s success!", tableNameStr);
            }
        } finally {
            if (admin != null) {
                admin.close();
            }
        }
    }


    /**
     * 插入数据
     * @param connection
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param data
     * @throws IOException
     */
    public static void put(Connection connection, TableName tableName,
                           String rowKey, String columnFamily, String column, String data) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            table.put(put);
        } finally {
            if (table != null) {
                table.close();
            }
        }
    }

    /**
     * 根据row key、column 读取
     * @param connection
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @return
     * @throws IOException
     */
    public static String getCell(Connection connection, TableName tableName, String rowKey, String columnFamily, String column) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column));

            Result result = table.get(get);
            List<Cell> cells = result.listCells();

            if (CollectionUtils.isEmpty(cells)) {
                return null;
            }
            String value = new String(CellUtil.cloneValue(cells.get(0)), "UTF-8");
            return value;
        } finally {
            if (table != null) {
                table.close();
            }
        }
    }

    /**
     * 一行数据
     * @param connection
     * @param tableName
     * @param rowKey
     * @return
     * @throws IOException
     */
    public static Map<String, String> getRow(Connection connection, TableName tableName, String rowKey) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));

            Result result = table.get(get);
            List<Cell> cells = result.listCells();

            if (CollectionUtils.isEmpty(cells)) {
                return Collections.emptyMap();
            }
            Map<String, String> objectMap = new HashMap<>();
            for (Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell), "UTF-8");
                objectMap.put(qualifier, value);
            }
            return objectMap;
        } finally {
            if (table != null) {
                table.close();
            }
        }
    }

    /**
     * 扫描全表
     * @param connection
     * @param tableName
     * @param rowkeyStart
     * @param rowkeyEnd
     * @return
     * @throws IOException
     */
    public static List<Map<String, String>> scan(Connection connection, TableName tableName, String rowkeyStart, String rowkeyEnd) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            ResultScanner rs = null;
            try {
                Scan scan = new Scan();
                if (!StringUtils.isEmpty(rowkeyStart)) {
                    scan.withStartRow(Bytes.toBytes(rowkeyStart));
                }
                if (!StringUtils.isEmpty(rowkeyEnd)) {
                    scan.withStopRow(Bytes.toBytes(rowkeyEnd));
                }
                rs = table.getScanner(scan);

                List<Map<String, String>> dataList = new ArrayList<>();
                for (Result r : rs) {
                    Map<String, String> objectMap = new HashMap<>();
                    for (Cell cell : r.listCells()) {
                        String qualifier = new String(CellUtil.cloneQualifier(cell));
                        String value = new String(CellUtil.cloneValue(cell), "UTF-8");
                        objectMap.put(qualifier, value);
                    }
                    dataList.add(objectMap);
                }
                return dataList;
            } finally {
                if (rs != null) {
                    rs.close();
                }
            }
        } finally {
            if (table != null) {
                table.close();
            }
        }
    }

    /**
     * 删表
     * @param connection
     * @param tableName
     * @throws IOException
     */
    public static void deleteTable(Connection connection, TableName tableName) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if (admin.tableExists(tableName)) {
                //现执行disable
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
        } finally {
            if (admin != null) {
                admin.close();
            }
        }
    }


    public static void main(String[] args) throws IOException {
        HbaseApiOperation hbaseApiOperation = new HbaseApiOperation();
        // 指定表的多个列簇
        String[] columnFamilys = {"info", "score"};
        TableName tableName = TableName.valueOf(tableNameStr);
        // 生成测试数据
//        ArrayList<Student>  studentArrayList = hbaseApiOperation.generateTestData();

        Connection connection = getConnection();
        // 1. 创建表操作
//        hbaseApiOperation.createTable(connection, tableName,columnFamilys,versions);
        // 2. 插入数据
//        put(connection, tableName, "Tom", "info", "student_id", "20210000000001");
//        put(connection, tableName, "Tom", "info", "class", "1");
//        put(connection, tableName, "Tom", "score", "understanding", "75");
//        put(connection, tableName, "Tom", "score", "programming", "82");
//
//        put(connection, tableName, "Jerry", "info", "student_id", "20210000000002");
//        put(connection, tableName, "Jerry", "info", "class", "1");
//        put(connection, tableName, "Jerry", "score", "understanding", "85");
//        put(connection, tableName, "Jerry", "score", "programming", "67");
//
//        put(connection, tableName, "Jack", "info", "student_id", "20210000000003");
//        put(connection, tableName, "Jack", "info", "class", "2");
//        put(connection, tableName, "Jack", "score", "understanding", "80");
//        put(connection, tableName, "Jack", "score", "programming", "80");
//
//        put(connection, tableName, "Rose", "info", "student_id", "20210000000004");
//        put(connection, tableName, "Rose", "info", "class", "2");
//        put(connection, tableName, "Rose", "score", "understanding", "60");
//        put(connection, tableName, "Rose", "score", "programming", "61");
//
//        put(connection, tableName, "HuJinhui", "info", "student_id", "20210000000011");
//        put(connection, tableName, "HuJinhui", "info", "class", "1");
//        put(connection, tableName, "HuJinhui", "score", "understanding", "99");
//        put(connection, tableName, "HuJinhui", "score", "programming", "99");

        // 3. 按照RowKey 查一行数据
        Map<String, String> row = getRow(connection, tableName, "Rose");
        System.out.printf("查一行数据: %s", row);
        System.out.println();
        // 4. 按照RowKey 查单元格数据
        String cell = getCell(connection, tableName, "Jack", "info", "student_id");
        System.out.printf("查单元格数据: %s", cell);
        System.out.println();
        // 5. 扫描全表数据
        List<Map<String, String>> dataList = scan(connection, tableName, null, null);
        System.out.printf("扫描表结果-: %s", dataList);
        System.out.println();
        // 6. 删除表
//        deleteTable(connection, tableName);
    }

}